package com.ibm.cps.processors;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import com.ibm.cps.dft.DFTOperator;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.message.ParsedDataMessageBuilder;
import com.ibm.cps.sql.expression.function.meta.FieldNameParam;
import com.ibm.cps.sql.expression.function.meta.IgnoredInFunctionCall;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.exception.CPSException;

// this processor is rewritten from DftDescriptor
@Info(Name=MessageFields.DFT_VECTOR_TYPE,Type=Info.ProcessorType.Vector)
public class DFTProcessor extends SchemaInheritiveProcessor{
	@FieldNameParam
	private String inputField;
	@FieldNameParam @IgnoredInFunctionCall
	private String outputField;

	public DFTProcessor(String tenantId, String processorId, String source, ITimeSeriesSchema parentSchema) throws CPSException {
		super(tenantId, processorId, source, parentSchema);
	}

	@Override
	protected List<NewField> defineFieldsToBeAdded() {
		NewField f = new NewField(outputField, "double[]");
		List<NewField> fields = new ArrayList<>();
		fields.add(f);
		return fields;
	}

	@Override
	public void dispose() {
	}

	

	@Override
	protected Collection<ParsedDataMessage> runProcessorLogic(ITimeSeriesSchema parentSchema,
			ITimeSeriesSchema outputSchema, ParsedDataMessage message) throws CPSException {
		Collection<ParsedDataMessage> messages = new ArrayList<>();
		ParsedDataMessageBuilder outputBuilder = ParsedDataMessageBuilder.createInhertiveBuilder(outputSchema, message, parentSchema);
		int index = parentSchema.getTagId(inputField);
		Double[] data = message.getArray(index);
		Double[] _data = new Double[data.length];
		for (int i = 0; i < data.length; i++) {
			_data[i] = (double) data[i];
		}
		DFTOperator operator = new DFTOperator();
		Double[] dftArr = operator.ComputDFTModulusVecotr(_data);
		outputBuilder.append(outputField, dftArr);
		messages.add(outputBuilder.toParsedDataMessage());
		return messages;
	}

	@Override
	public void initialize() throws CPSException {
		
	}

}